package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 1、创建flink的运行环境
      * 是flink程序的入口
      *
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置并行度, 默认并行度是计算机核心数
    env.setParallelism(2)

    /**
      * 2、读取数据
      *
      * DataStream ： 相当于spark中DStream
      *
      *
      * 开启scoket
      * nc -lk 8888
      *
      */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    //1、将数据展开
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))


    //2、转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))

    //3、按照单词进行分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)
    /**
      *
      * flink中的算子本身就是有状态的算子，
      *
      */

    //4、统计数量,对value进行求和, 指定下标进行聚合
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)



    //打印结果
    countDS.print()

    /**
      * 启动flink程序
      *
      */

    env.execute("wordcount")

  }

}
